/*
 Copyright 2009 Abdulla Abdurakhmanov (abdulla@latestbit.com)
 Original sources are available at www.latestbit.com

 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
 You may obtain a copy of the License at

 http://www.apache.org/licenses/LICENSE-2.0

 Unless required by applicable law or agreed to in writing, software
 distributed under the License is distributed on an "AS IS" BASIS,
 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.

 ServerSkeleton.cpp. Created on: 28.10.2009 15:32:47 
 */

#include "smartobjects/core/ServerSkeleton.hpp"
#include "smartobjects/coders/CallInterfaceOperationRequest.hpp"

namespace smartobjects {

ServerSkeleton::ServerSkeleton()
		: m_nRequestQueuePerClientIdx(0),
		  m_poThreadPool(NULL),
		  m_nTaskId(0),
		  m_nThreadModel(IServerSkeleton::SINGLE_THREAD_MODEL),
		  m_poEndpointLocation(NULL),
		  m_bRegistered(false), m_bPublished(false), m_bDefferedUnpublish(false),
		  m_poAvailForConnectionHandler(NULL) {
}

ServerSkeleton::~ServerSkeleton() {
	releaseCurrentNetworkEnv();
}

void ServerSkeleton::releaseCurrentNetworkEnv() {
	boost::unique_lock<boost::recursive_mutex> oLock(m_mxSkeleton);

	if(m_poTlsCallProcessingStatus.get()!=NULL && *m_poTlsCallProcessingStatus) {
		m_bDefferedUnpublish = true;
		return;
	}
	m_bDefferedUnpublish = false;

	m_bPublished = false;
	if(m_poThreadPool) {
		m_poThreadPool->cancel(m_nTaskId);
		m_poThreadPool->wait(m_nTaskId);
	}
	releaseAndClear(m_poEndpointLocation);
	if(m_poBusServer) {
		ICommandProcessor* poCommandProcessor = m_poBusServer->getCommandProcessor();
		if(poCommandProcessor) {
			poCommandProcessor->delRequestHandler(this);
		}
		SmartBusServer* poServer = m_poBusServer;

		boost::unique_lock<boost::shared_mutex> oRequestProcessorLock(m_mxRequestQueueProcessor);
		boost::unique_lock<boost::recursive_mutex> oRequestQueueLock(m_mxRequestQueue);
		m_poBusServer = NULL;
		m_aRequestQueue.clear();
		m_mRequestQueuePerClient.clear();
		oRequestQueueLock.unlock();
		oRequestProcessorLock.unlock();

		poServer->release();
	}
	else
	if(m_poBusClient) {
		ICommandProcessor* poCommandProcessor = m_poBusClient->getCommandProcessor();
		if(poCommandProcessor) {
			poCommandProcessor->delRequestHandler(this);
		}
		SmartBusClient* poClient = m_poBusClient;

		boost::unique_lock<boost::shared_mutex> oRequestProcessorLock(m_mxRequestQueueProcessor);
		boost::unique_lock<boost::recursive_mutex> oRequestQueueLock(m_mxRequestQueue);
		m_poBusClient = NULL;
		m_aRequestQueue.clear();
		m_mRequestQueuePerClient.clear();
		oRequestQueueLock.unlock();
		oRequestProcessorLock.unlock();

		poClient->release();
	}

	if(m_bRegistered) {
		unregisterInstance();
	}

	m_poThreadPool = NULL;
	if(m_poBus) {
		setBus(NULL);
	}
}

void ServerSkeleton::unpublish() {
	releaseCurrentNetworkEnv();
}

void ServerSkeleton::publish( ISmartBus* poBus, IEndpointLocation* poLocation) {
	boost::unique_lock<boost::recursive_mutex> oLock(m_mxSkeleton);
	try {
		if(m_poBusClient || m_poBusServer) {
			releaseCurrentNetworkEnv();
		}
		setBus(poBus);
		setInstanceId("");
		m_poBusServer = m_poBus->selectBusServer(poLocation);
		m_poThreadPool = m_poBus->getThreadPool();
		m_nTaskId = m_poThreadPool->createTaskId();

		m_poBusServer->getCommandProcessor()->addRequestHandler(this, this);
		m_poBusServer->start();
		setNewRefObject(m_poEndpointLocation, poLocation);
		m_bPublished = true;
	}
	catch(const cpputils::IOException& ex) {
		throw SmartBusSystemException(ex.what());
	}
}

void ServerSkeleton::publish( ISmartObject* poBaseObject ) {
	boost::unique_lock<boost::recursive_mutex> oLock(m_mxSkeleton);

	if(!poBaseObject->getBusClient() && !poBaseObject->getBusServer()) {
		throw SmartBusSystemException("Unable to publish object on unpublished/resolved base object!");
	}

	if(m_poBusClient || m_poBusServer) {
		releaseCurrentNetworkEnv();
	}
	setBus(poBaseObject->getBus());
	setInstanceId( (boost::format("%1%") %this).str());

	m_poThreadPool = m_poBus->getThreadPool();
	m_nTaskId = m_poThreadPool->createTaskId();

	if(poBaseObject->getBusServer()) {
		m_poBusServer = poBaseObject->getBusServer();
		m_poBusServer->addRef();
		m_poBusServer->getCommandProcessor()->addRequestHandler(this, this);
	}
	else {
		m_poBusClient = poBaseObject->getBusClient();
		m_poBusClient->addRef();
		m_poBusClient->getCommandProcessor()->addRequestHandler(this, this);
	}

	m_bPublished = true;
}

void ServerSkeleton::publish( ISmartObject* poBaseObject, ConnectionHandler* poConnectionHandler ) {
	boost::unique_lock<boost::recursive_mutex> oLock(m_mxSkeleton);
	publish(poBaseObject);
	registerInstance(poConnectionHandler);
}

void ServerSkeleton::registerInstance ( ConnectionHandler* poConnectionHandler ) {
	boost::unique_lock<boost::recursive_mutex> oLock(m_mxSkeleton);
	IServerInstancesRegistry* poRegistry = m_poBus->getServerInstancesRegistry();
	poRegistry->addInstance( this );
	poRegistry->attachInstanceClientHandler( poConnectionHandler, this );
	m_bRegistered = true;
	if(!m_oLifecycleControlParams.isAvailableByInstanceIdForAll()) {
		// Non reference copy because the pointer saved in m_poAvailForConnectionHandler is not used to access
		// for read/write it members
		m_poAvailForConnectionHandler = poConnectionHandler;
	}
	else
		m_poAvailForConnectionHandler = NULL;
}

void ServerSkeleton::unregisterInstance () {
	boost::unique_lock<boost::recursive_mutex> oLock(m_mxSkeleton);
	if(m_bRegistered) {
		IServerInstancesRegistry* poRegistry = m_poBus->getServerInstancesRegistry();
		poRegistry->removeInstance(this);
		m_bRegistered = false;
		m_poAvailForConnectionHandler = NULL;
	}
}

LifecycleControlParams& ServerSkeleton::getLifecycleControlParams() {
	return m_oLifecycleControlParams;
}

bool ServerSkeleton::isPublished() const {
	return m_bPublished;
}

IEndpointLocation* ServerSkeleton::getPublishedLocation() {
	boost::unique_lock<boost::recursive_mutex> oLock(m_mxSkeleton);
	return m_poEndpointLocation;
}

void ServerSkeleton::setThreadModel(ThreadModel nModel) {
	boost::unique_lock<boost::recursive_mutex> oLock(m_mxSkeleton);
	m_nThreadModel = nModel;
}

IServerSkeleton::ThreadModel ServerSkeleton::getThreadModel() {
	boost::unique_lock<boost::recursive_mutex> oLock(m_mxSkeleton);
	return m_nThreadModel;
}

void ServerSkeleton::onResponseReceived(
		ConnectionHandler* poHandler,
		INetworkCommand* poRequest,
		INetworkCommand* poResponse,
		RefObject* poUserObject) {
	// Nothing to do - server never receive response
}

void ServerSkeleton::onResponseTimeout(
		INetworkCommand* poRequest,
		RefObject* poUserObject) {
	// Nothing to do - server never receive response
}

bool ServerSkeleton::isCommandIdSupported(
		ConnectionHandler* poHandler,
		INetworkCommand* poRequest) {
	if( poRequest->getCommandId() == CallInterfaceOperationRequest::COMMAND_ID ) {
		if( m_poAvailForConnectionHandler != NULL && poHandler != m_poAvailForConnectionHandler ) {
			return false;
		}
		else
			return true;
	}
	else
		return false;
}

void ServerSkeleton::onRequestReceived(
		ConnectionHandler* poHandler,
		INetworkCommand* poRequest) {

	if(!isPublished()) {
		return;
	}

	updateLastActivity();

	RequestQueueItem oItem;
	oItem.setConnectionHandler(poHandler);
	oItem.setRequest(poRequest);

	bool bIsNeedToRunTask = true;

	boost::unique_lock<boost::recursive_mutex> oRequestQueueLock(m_mxRequestQueue);
	switch( m_nThreadModel ) {
		case IServerSkeleton::SINGLE_THREAD_MODEL:
		{
			if(!m_aRequestQueue.empty()) {
				bIsNeedToRunTask = false;
			}
			m_aRequestQueue.push_back(oItem);
			break;
		}
		case IServerSkeleton::PER_CLIENT_MULTI_THREAD_MODEL:
		{
			RequestQueuePerClient::iterator it = m_mRequestQueuePerClient.find(poHandler);
			RequestQueue& aQueuePerClient = m_mRequestQueuePerClient[poHandler];
			if(!aQueuePerClient.empty()) {
				bIsNeedToRunTask = false;
			}
			aQueuePerClient.push_back(oItem);
			break;
		}
		case IServerSkeleton::FULL_MULTI_THREAD_MODEL:
		{
			m_aRequestQueue.push_back(oItem);
			break;
		}
	}
	oRequestQueueLock.unlock();

	if(bIsNeedToRunTask) {
		m_poThreadPool->schedule(
				m_nTaskId,
				boost::bind(&ServerSkeleton::onDequeRequest, this)
		);
	}
}

void ServerSkeleton::onDequeRequest() {
	RequestQueueItem oItem;
	RequestQueue* poQueuePerClient = NULL;

	boost::shared_lock<boost::shared_mutex> oRequestProcessorLock(m_mxRequestQueueProcessor);
	if(m_nThreadModel == IServerSkeleton::PER_CLIENT_MULTI_THREAD_MODEL) {
		boost::unique_lock<boost::recursive_mutex> oRequestQueueLock(m_mxRequestQueue);
		if(!m_mRequestQueuePerClient.empty()) {
			if(m_mRequestQueuePerClient.size()<=m_nRequestQueuePerClientIdx)
				throw SystemException("Internal error! Queue per client idx is wrong!");

			RequestQueuePerClient::iterator itRQPC = m_mRequestQueuePerClient.begin();
			for(unsigned int nIdx = 0; nIdx < m_nRequestQueuePerClientIdx; nIdx++) {
				itRQPC++;
			}
			m_nRequestQueuePerClientIdx++;
			poQueuePerClient = &itRQPC->second;
		}
	}

	do {
		boost::unique_lock<boost::recursive_mutex> oRequestQueueLock(m_mxRequestQueue);
		oItem.setRequest(NULL);
		if(m_nThreadModel != IServerSkeleton::PER_CLIENT_MULTI_THREAD_MODEL) {
			oItem.setConnectionHandler(NULL);
			if(!m_aRequestQueue.empty()) {
				oItem = m_aRequestQueue.front();
				m_aRequestQueue.pop_front();
			}
		}
		else {
			if(poQueuePerClient && !poQueuePerClient->empty()) {
				oItem = poQueuePerClient->front();
				poQueuePerClient->pop_front();
			}
		}
		oRequestQueueLock.unlock();

		if(!m_poBusServer && !m_poBusClient) {
			break;
		}

		if(oItem.getRequest()!=NULL) {
			boost::intrusive_ptr<CallInterfaceOperationRequest> poRequest (
					new CallInterfaceOperationRequest(oItem.getRequest()), false
			);
			try {
				this->addRef();
				poRequest->deserialize();
				if(!poRequest->getRequestContext()) {
					boost::intrusive_ptr<RequestContext> poNewContext ( new RequestContext(), false );
					poRequest->setRequestContext( poNewContext.get() );
				}
				poRequest->getRequestContext()->setBus(m_poBus);
				poRequest->getRequestContext()->setConnectionHandler(oItem.getConnectionHandler());
				poRequest->getRequestContext()->setSkeleton(this);

				try {
					if(m_poTlsCallProcessingStatus.get() == NULL)
						m_poTlsCallProcessingStatus.reset( new bool() );
					*m_poTlsCallProcessingStatus = true;
					poRequest->getRequestContext()->addRef();
					m_poTlsCurrentRequestContext.reset ( poRequest->getRequestContext() );
					this->onCallRequestReceived(oItem.getConnectionHandler(), poRequest.get());
				}
				catch(const SmartBusSystemException& ex) {
					poRequest->getBuffer()->clear();
					try {
						sendSystemExceptionResponse(oItem.getConnectionHandler(), poRequest.get(), ex);
					}
					catch(const std::exception& ex) {
						std::cout << ex.what() << std::endl;
					}
				}
				m_poTlsCallProcessingStatus.reset();
				m_poTlsCurrentRequestContext.reset();

				poRequest->getRequestContext()->setSkeleton(NULL);
				oRequestProcessorLock.unlock();
				if(m_bDefferedUnpublish) {
					this->addRef();
					unpublish();
					if(this->release())
						return;
				}
				if(this->release())
					return;
				oRequestProcessorLock.lock();
			}
			catch(const std::exception& ex) {
				std::cout << ex.what() << std::endl;
				oRequestProcessorLock.unlock();
				if(this->release())
					return;
				oRequestProcessorLock.lock();
			}
		}
		else {
			if(m_nThreadModel == IServerSkeleton::PER_CLIENT_MULTI_THREAD_MODEL) {
				boost::unique_lock<boost::recursive_mutex> oRequestQueueLock(m_mxRequestQueue);
				if(oItem.getConnectionHandler()!=NULL) {
					m_mRequestQueuePerClient.erase(oItem.getConnectionHandler());
					m_nRequestQueuePerClientIdx--;
					oItem.setConnectionHandler(NULL);
				}
			}
		}
	}
	while(oItem.getRequest());
}

void ServerSkeleton::sendResponse(ConnectionHandler* poConnectionHandler, CallInterfaceOperationResponse* poResponse) {
	ICommandProcessor* poProcessor = getCurrentCommandProcessor();
	if(poProcessor) {
		poProcessor->sendResponse(poConnectionHandler, poResponse);
	}
}

void ServerSkeleton::sendSystemExceptionResponse(ConnectionHandler* poConnectionHandler, CallInterfaceOperationRequest* poRequest, const std::exception& ex) {
	boost::intrusive_ptr<CallInterfaceOperationResponse> poResponse (
			poRequest->createResponse(poRequest->getBuffer()),
			false
	);
	poRequest->setBuffer(NULL);
	poResponse->setExceptionFlag(true);
	poResponse->setExceptionInterfaceId("SmartBusSystemException");
	poResponse->setExceptionMessage(ex.what());
	poResponse->beginSerialize();
	poResponse->finishSerialize();
	sendResponse(poConnectionHandler, poResponse.get());
}

ICommandProcessor* ServerSkeleton::getCurrentCommandProcessor() {
	ICommandProcessor* poResult = NULL;
	if(m_poBusServer) {
		poResult = m_poBusServer->getCommandProcessor();
	}
	else
	if(m_poBusClient) {
		poResult = m_poBusClient->getCommandProcessor();
	}
	return poResult;
}

size_t ServerSkeleton::getRequestQueueSize() const {
	size_t nResult = 0;
	boost::unique_lock<boost::recursive_mutex> oRequestQueueLock(m_mxRequestQueue);
	switch( m_nThreadModel ) {
		case IServerSkeleton::SINGLE_THREAD_MODEL:
		case IServerSkeleton::FULL_MULTI_THREAD_MODEL:
		{
			nResult = m_aRequestQueue.size();
			break;
		}
		case IServerSkeleton::PER_CLIENT_MULTI_THREAD_MODEL:
		{
			for(RequestQueuePerClient::const_iterator itClient = m_mRequestQueuePerClient.begin();
					itClient!=m_mRequestQueuePerClient.end();
					itClient++ ) {
				nResult+=itClient->second.size();
			}
			break;
		}
	}
	return nResult;
}

boost::thread_specific_ptr<RefObject> ServerSkeleton::m_poTlsCurrentRequestContext(&boost::intrusive_ptr_release);

RequestContext* ServerSkeleton::getCurrentRequestContext() {
	return dynamic_cast<RequestContext*> ( m_poTlsCurrentRequestContext.get());
}

}
